package org.raymond.iworks.core.serializer;

import lombok.extern.slf4j.Slf4j;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.common.serialization.Deserializer;
import org.raymond.iworks.commons.constants.MQConstants;
import org.raymond.iworks.apis.avro.LogModel;

import java.util.HashMap;
import java.util.Map;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class KafkaAvroDeserializer<T extends SpecificRecordBase> implements Deserializer<T> {
    private static final Map<String, Class> typeMap = new HashMap<>();

    static {
        typeMap.put(MQConstants.TOPIC_LOG, LogModel.class);
        typeMap.put("demo3", LogModel.class);
    }

    @Override
    public T deserialize(String topic, byte[] bytes) {
        Class clazz = typeMap.get(topic);
        DatumReader<T> reader = new SpecificDatumReader<>(clazz);
        BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
        T t = null;
        try {
            t = reader.read(null, decoder);
        }catch (Exception e){
            log.error("{}", e.getMessage());
        }
        return t;
    }
}
